<?php

namespace Core\Server;

use Core\Common\Argus\Interceptor\ArgusInterceptor;
use Core\Common\Base\Config;
use Core\Common\Base\ConfigSource;
use Core\Common\Base\GameWebSocket;
use Core\Common\Base\Job;
use Core\Common\Base\Lock;
use Core\Common\Base\Log;
use Core\Common\Base\Process;
use Core\Common\Base\Service;
use Core\Common\Base\Watcher;
use Core\Common\Base\WebSocket;
use Core\Common\Codec\Redis as RedisCodec;
use Core\Common\Route\Dispatcher;
use Core\Common\Route\Request;
use Core\Common\Route\Response;
use Core\Common\Util\DbError;
use Core\Common\Util\Helper;
use Swoole\Http\Request as SwooleRequest;
use Swoole\Http\Response as SwooleResponse;
use Swoole\Redis\Server;

/**
 * 集成Http协议, Redis协议, WebSocket协议三种协议的服务端
 * Description of SuperServer
 *
 * @author baiyang <baiyang@ibantang.com>
 */
abstract class SuperServer
{
    /**
     * @var string
     */
    protected $host = '0.0.0.0';

    /**
     * @var int
     */
    protected $port = 8084;

    /**
     * @var array
     */
    protected $services = [];

    /**
     * @var null
     * 主服务
     */
    public $instance = null;

    /**
     * @var null
     * Redis协议服务
     */
    protected $redisService = null;

    /**
     * @var \Core\Common\Base\GameWebSocket
     * websocket协议服务
     */
    protected $webSocketService = null;

    /**
     * @var null
     */
    protected $currentFd = null;

    /**
     * @var \Core\Common\Codec\Redis
     */
    protected $redisCodec = null;

    /**
     * @var \Swoole\Table
     */
    protected $routes = [];

    /**
     * @var \Swoole\Table
     */
    protected $routeMiddlewares = [];

    /**
     * 由于DB连接错误报出的错误信息
     * @var array
     */
    protected $lostDBConnectionErrorMsgs = [];

    /**
     * 从consul里加载的配置文件信息
     * @var \Swoole\Table
     */
    protected $appConfig = null;

    /**
     * SuperServer constructor.
     * @param array $services
     */
    public function __construct($services = [])
    {
        $this->services = $services;
    }

    /**
     *
     */
    public function __destruct()
    {
        $this->instance = null;
        $this->redisService = null;
        $this->webSocketService = null;
    }

    /**
     * @param $serv
     */
    public function onStart($serv)
    {
        if ('Linux' == PHP_OS) {
            swoole_set_process_name(CORE_NAME . 'Service:Master Process');
        }
        file_put_contents(CORE_PID_DIR . 'master', $serv->master_pid);
    }

    /**
     * @param $serv
     */
    public function onManagerStart($serv)
    {
        if ('Linux' == PHP_OS) {
            swoole_set_process_name(CORE_NAME . 'Service:Manager Process');
        }
        file_put_contents(CORE_PID_DIR . 'manager', $serv->manager_pid);
    }

    /**
     * @param $serv
     * @param $worker_id
     */
    public function onWorkerStart($serv, $worker_id)
    {
        $this->beforeWorkerStart();
        if (function_exists('opcache_reset')) {
            opcache_reset();
        }
        if (function_exists('apc_clear_cache')) {
            apc_clear_cache();
        }
        Config::init();
        if ($worker_id >= $serv->setting['worker_num']) {
            if ('Linux' == PHP_OS) {
                swoole_set_process_name(CORE_NAME . 'Service:Task Process');
            }
            define('CORE_PROCESS', 'task');
        } else {
            if ('Linux' == PHP_OS) {
                swoole_set_process_name(CORE_NAME . 'Service:Worker Process');
            }
            define('CORE_PROCESS', 'worker');
            if ($this->routes->count()) {
                Dispatcher::getInstance(CORE_WORK_ROOT)->setUriObj($this->routes);
            }
        }
        if ($worker_id == $serv->setting['worker_num']) {
            $routes = Dispatcher::getInstance(CORE_WORK_ROOT)->initUri();
            foreach ($routes as $k => $v) {
                $this->routes->del($k);
            }
            foreach ($routes as $k => $v) {
                $this->routes->set($k, ['detail' => json_encode($v)]);
            }
            for ($i = 0; $i < $serv->setting['worker_num']; ++$i) {
                $serv->sendMessage('route', $i);
            }
        }
        /**
         * 注册异步日志处理逻辑
         */
        Watcher::getInstance()->register('log', function ($name, $msg, $prefix = 'flm_') use ($serv) {
            if (defined('CORE_PROCESS') && CORE_PROCESS == 'worker') {
                $serv->task(['log', $name, $msg, $prefix]);
            } else {
                Log::save($name, $msg, $prefix);
            }
        });
        Watcher::getInstance()->register('job', function (...$args) use ($serv) {
            $serv->task(['job', $args]);
        });
        static::registerWatcher();
        $this->onInit();
    }

    /**
     * 处理日志，及其他Job
     * @param $serv
     * @param int $task_id
     * @param int $src_worker_id
     * @param $data
     */
    public function onTask($serv, int $task_id, int $src_worker_id, $data)
    {
        $type = $data[0];
        switch ($type) {
            case 'log'://异步写日志
                //list(, $name, $str) = $data;
                array_shift($data);
                Log::save(...$data);
                break;
            case 'job'://处理任务
                $args = $data[1];
                Job::execute(...$args);
                break;
            case 'cast'://发到指定的worker, ['cast', $message, $taskId]
                $serv->sendMessage($data[1], $data[2]);
                break;
            case '/websocket/pushAll'://发到指定的worker, ['cast', $message, $taskId]
                $this->webSocketService->pushAll($data[1]);
                break;
            default:
                break;
        }
    }

    /**
     * @param $serv
     * @param int $task_id
     * @param string $data
     */
    public function onFinish($serv, int $task_id, string $data)
    {
    }

    /**
     * @param $serv
     * @param $src_worker_id
     * @param $data
     */
    public function onPipeMessage($serv, $src_worker_id, $data)
    {
        if (is_string($data) && 'route' == $data && $this->routes->count()) {//处理路由
            Dispatcher::getInstance(CORE_WORK_ROOT)->setUriObj($this->routes);
            return;
        }
//        if (is_array($data) && isset($data[0]) && 'cast' == $data[0]) {
//            if ('task' == CORE_PROCESS) {
//                list (, $message, $workerId) = $data;
//                if ($serv->worker_id == $workerId) {
//                    $workerId = ($workerId == 0) ? $workerId + 1 : $workerId - 1;
//                }
//                $serv->sendMessage($message, $workerId);
//            }
//            return;
//        }
//        Log::pipe(LOG_INFO, ['message' => $data]);
//        if (!class_exists('App\Consumer\CommonConsumer')) {
//            Log::error(LOG_INFO, ['error' => "App\Consumer\CommonConsumer not found", ]);
//            return;
//        }
//        call_user_func(['App\Consumer\CommonConsumer', 'run'], $data);
    }

    public function dispatch($uri, $param = [])
    {
        if (Request::isHttp2()) {
            Dispatcher::getInstance(CORE_WORK_ROOT)->http2(trim($uri, '/'));
        } else {
            Dispatcher::getInstance(CORE_WORK_ROOT)->proceed(strtolower(trim($uri, '/')), $param);
        }
    }

    /**
     * 服务器统计信息
     */
    public function stats()
    {
        $data = [
            'connections' => count($this->instance->connections),
            'lastError' => $this->instance->getLastError(),
            'server_setting' => $this->instance->setting,
            'stats' => $this->instance->stats(),
            'name' => CORE_NAME,
            'env' => CORE_ENV,
            'root_path' => CORE_WORK_ROOT,
            'config_path' => CORE_CONFIG_PATH,
            'log_path' => CORE_LOG_PATH,
            'memory_used' => (memory_get_usage(true) / 1024 / 1024) . 'MB',
            'memory_used_peak' => (memory_get_peak_usage(true) / 1024 / 1024) . 'MB',
            'services' => implode(',', $this->services),
            'app_config_reset_time' => ConfigSource::getResetAppConfigTime(),
            'service_gateway_route' => Service::getServiceGatewayRouteInfo(),
        ];
        Response::succ($data + $this->extraStats());
    }

    public function extraStats()
    {
        return [];
    }

    /**
     * 处理HTTP请求
     * @param SwooleRequest $request
     * @param SwooleResponse $response
     */
    public function onHttpRequest(SwooleRequest $request, SwooleResponse $response)
    {
        if (isset($request->header['user-agent']) && false !== (strpos($request->header['user-agent'], 'Consul'))) {
            $response->status(200);
            $response->end('OK');
            return;
        }
        if ('/healthy_check' == $request->server['request_uri']) {
            // 针对健康检查的响应
            $response->status(200);
            $response->end('OK');
            return;
        }
        if ('/favicon.ico' == $request->server['request_uri']) {
            $response->status(404);
            $response->end();
            return;
        }
        $this->currentFd = $response->fd;
        Request::reset();
        Response::reset();
        Request::setHttp();

        foreach ($request->server as $k => $v) {
            Request::server(strtoupper($k), $v);
        }
        foreach ($request->header as $k => $v) {
            //Request::server(strtoupper($k), $v);
            $nk = 'HTTP_' . strtoupper(str_replace('-', '_', $k));
            Request::server($nk, $v);
            $_SERVER[$nk] = $v;
        }
        foreach ((array)$request->get as $k => $v) {
            Request::get($k, $v);
        }
        foreach ((array)$request->post as $k => $v) {
            Request::post($k, $v);
        }
        foreach ((array)$request->cookie as $k => $v) {
            Request::cookie($k, $v);
        }
        if (!isset($request->header['content-type']) || false === strpos($request->header['content-type'], 'form')) {
            Request::raw($request->rawContent());
        }
        $t1 = (string)microtime(true);
        $logLevel = LOG_NOTICE;

        if (Request::isArgusSimple()) {
            ArgusInterceptor::handleRequest('sr');
        }

        $isInterrupt = $this->interruptRequest();
        if (true === $isInterrupt) {
            $response->end("");
            Response::reset();
            Request::reset();
            $_SERVER = null;
            return;
        }

        $this->beforeRequest();
        try {
            $uri = trim(Request::server('REQUEST_URI'), '/');
            if ('_stats' == $uri) {
                $this->stats();
            } else {
                if ('_reload' == $uri) {
                    $this->instance->reload();
                    Response::succ('Reload OK');
                } else {
                    if ('_wspush' == $uri) {
                        $this->webSocketService->pushActionHandler();
                    } else {
                        $this->dispatch($uri);
                    }
                }
            }
        } catch (\Throwable $ex) {
            Log::exception(
                LOG_ERR,
                [
                    'http_code' => Response::httpCode(),
                    'file' => $ex->getFile() . ':' . $ex->getLine(),
                    'message' => $ex->getMessage(),
                    'trace' => $ex->getTrace()
                ]
            );
            if (Request::inDebug()) {
                Response::debug($ex->getMessage() . "<br/><pre>" . $ex->getTraceAsString() . '</pre>');
            }
            Lock::unlock();
            $logLevel = LOG_ERR;
        }
        $code = Response::httpCode();
        if ($code > 1) {
            $response->status($code);
        }
        $str = Response::result();
        Log::access($logLevel, [
            'http_code' => $code,
            'response_size' => strlen($str),
            'ms' => (microtime(true) - $t1) * 1000,
        ]);
        if (!Request::isGrpc()) {
            $response->header('hn', Helper::getHostName());
            $response->header('Content-Type', Response::contentType());
            if (ConfigSource::isUseConsul()) {
                $response->header('X-Config-Reset-Time', ConfigSource::getResetAppConfigTime());
            }
            if (version_compare(SWOOLE_VERSION, '4.1.0') < 0) {
                $acceptEncoding = Request::server('HTTP_ACCEPT_ENCODING');
                if (strlen($str) > 10240
                    && $acceptEncoding
                    && (false !== strpos($acceptEncoding, 'gzip')) || false !== strpos($acceptEncoding, '*/*')) {
                    $response->header('Content-Encoding', 'gzip');
                    $str = gzencode($str, 1);
                }
            }
            $cookies = Response::cookie();
            if ($cookies) {
                foreach ($cookies as $v) {
                    $response->rawCookie(...$v);
                }
            }
        }
        if (Request::isArgusSimple()) {
            ArgusInterceptor::handleRequest('ss');
        }
        $headers = Response::header();
        if ($headers) {
            foreach ($headers as $k => $v) {
                $response->header($k, $v);
            }
        }
        $this->afterRequest();
        $response->end($str);

        // 记录response内容
        if (Config::config('base.responseLog')
            && defined('\App\Config\RecordResponseUri::URI_LIST')
            && in_array(strtolower($uri), \App\Config\RecordResponseUri::URI_LIST)
        ) {
            Log::response($logLevel, [
                'response' => $str,
            ]);
        }

        Response::reset();
        Request::reset();
        $_SERVER = null;
    }

    /**
     * 处理redis协议的请求
     * @param $serv
     * @param $fd
     * @param $reactorId
     * @param $data
     */
    public function onRedisRequest($serv, $fd, $reactorId, $data)
    {
        $tmp = $this->redisCodec->unserialize($data);
        Response::reset();
        Request::reset();
        Request::setRedis();
        $this->currentFd = $fd;
        $t1 = microtime(true);
        try {
            $cmd = array_shift($tmp[0]);
            Request::setCmd($cmd);
            $this->dispatch($cmd, $tmp[0]);
        } catch (\Throwable $ex) {
            Log::exception(
                LOG_ERR,
                ['http_code' => Response::httpCode(), 'message' => $ex->getMessage(), 'trace' => $ex->getTrace()]
            );
            $serv->send($fd, $this->redisCodec->serialize($ex));
            Log::access(LOG_ERR, [
                'code' => Response::getCode(),
                'ms' => (microtime(true) - $t1) * 1000,
            ]);
            return;
        }
        Log::access(LOG_NOTICE, [
            'code' => Response::getCode(),
            'ms' => (microtime(true) - $t1) * 1000,
        ]);
        if (Response::getCode() > 1) {
            $serv->send($fd, $this->redisCodec->serialize('-' . json_encode(Response::rawResult())));
            return;
        }
        $serv->send($fd, $this->redisCodec->serialize(Response::data()));
        Request::reset();
        Response::reset();
    }

    /**
     * 获取框架启动所需要的swoole配置
     * @return array|bool
     */
    public function getService()
    {
        return Config::getSwooleServerConfig();
    }

    /**
     * 启动服务
     */
    public function start()
    {
        $this->routes = new \Swoole\Table(2000);
        $this->routes->column('detail', \Swoole\Table::TYPE_STRING, 300);
        $this->routes->create();

        $this->routeMiddlewares = new \Swoole\Table(1000);
        $this->routeMiddlewares->column('middleware', \Swoole\Table::TYPE_STRING, 100);
        $this->routeMiddlewares->create();

        register_shutdown_function([$this, 'onShutdown']);
        set_error_handler([$this, 'onError']);
        Lock::initLockTable();

        Service::serviceBootstrap();

        // 加载项目配置文件
        if (ConfigSource::isUseConsul()) {
            $configInitRet = ConfigSource::initTableAndInitConfig();
            if (!$configInitRet["ok"]) {
                Log::coreLog([
                    "msg" => "fetch appConfig error, " . $configInitRet["msg"],
                    "time" => date("Y-m-d H:i:s"),
                ]);
                exit();
            } else {
                $this->appConfig = $configInitRet["table"];
            }
        }

        $this->beforeStart();
        $config = [
            'reactor_num' => 2,
            'worker_num' => 16,
            'task_worker_num' => 2,
            'max_request' => 5000,
            'max_conn' => 65535,
            'daemonize' => true,
            'dispatch_mode' => 1,
            'open_tcp_keepalive' => 1,
            'document_root' => CORE_WORK_ROOT,
            //'enable_static_handler' => true,
            'request_slowlog_timeout' => 1,
            'trace_event_worker' => true,
            //'open_http2_protocol' => true,
            'http_compression' => true,
            'http_compression_level' => true,
        ];
        $service = (array)Config::getSwooleServerConfig();
        foreach ($service as $k => $v) {
            if ('app_log' == $k && !empty($v)) {
                define('CORE_LOG_PATH', rtrim($v, DIRECTORY_SEPARATOR) . DIRECTORY_SEPARATOR);
                continue;
            }
            $config[$k] = $v;
        }
        if (!defined('CORE_LOG_PATH')) {
            define('CORE_LOG_PATH', '/tmp/');
        }
        if (!file_exists(CORE_LOG_PATH)) {
            mkdir(CORE_LOG_PATH);
        }
        $config['log_file'] = CORE_LOG_PATH . CORE_NAME . '.log';
        $config['request_slowlog_file'] = CORE_LOG_PATH . CORE_NAME . '-slow.log';
        if (isset($config['host']) && $config['host']) {
            $this->host = $config['host'];
        }
        if (isset($config['port']) && $config['port']) {
            $this->port = $config['port'];
        }
        if (in_array('websocket', $this->services)) {
            $this->instance = new \Swoole\WebSocket\Server($this->host, $this->port + 2);
            $this->webSocketService = $this->registerWsService();

            //启用心跳检测,设置心在跳间隔和连接最大允许空闲的时间
            $config['heartbeat_idle_time'] = $config['heartbeat_idle_time'] ?? 60;
            $config['heartbeat_check_interval'] = $config['heartbeat_check_interval'] ?? 60;
            //设置websocket运行在固定模式
            $config['dispatch_mode'] = 2;

            $this->instance->on('open', [$this->webSocketService, 'onWsOpen']);
            $this->instance->on('message', [$this->webSocketService, 'onWsRequest']);
            $this->instance->on('close', [$this->webSocketService, 'onWsClose']);
        } else {
            $this->instance = new \Swoole\Http\Server($this->host, $this->port);
        }
        $this->instance->on('start', [$this, 'onStart']);
        $this->instance->on('managerStart', [$this, 'onManagerStart']);
        $this->instance->on('WorkerStart', [$this, 'onWorkerStart']);
        $this->instance->on('request', [$this, 'onHttpRequest']);
        $this->instance->on('task', [$this, 'onTask']);
        $this->instance->on('finish', [$this, 'onFinish']);
        $this->instance->on('pipeMessage', [$this, 'onPipeMessage']);
        $this->instance->on('shutdown', [$this, 'onServerShutdown']);
        if (in_array('redis', $this->services)) {
            $this->redisCodec = new RedisCodec();
            $redisPort = $this->instance->listen($this->host, $this->port + 1, SWOOLE_SOCK_TCP);
            $redisPort->set([
                'open_redis_protocol' => true,
                'open_http_protocol' => false,
            ]);
            $redisPort->on('receive', [$this, 'onRedisRequest']);
            $this->redisService = $redisPort;
        }
        $this->instance->set($config);
        $this->initProcess($config);
        $this->instance->start();
    }

    //自定义进程处理
    private function initProcess($config)
    {
        if (!file_exists(CORE_PID_DIR)) {
            mkdir(CORE_PID_DIR);
        }

        $processes = [];

        // 强制开启主进程监控
        $processes['monitor'] = 1;

        // 开启服务发现进程
        $processes['service'] = 1;

        $process = new Process($this, 'monitor', 1, 'monitor');
        $this->instance->addProcess(new \swoole_process([$process, 'run']));

        $processService = new Process($this, 'consulservice', 1, 'service');
        $this->instance->addProcess(new \swoole_process([$processService, 'run']));

        //多个自定义进程配置
        if (!empty($config['process_config'])) {
            $processArr = explode(',', $config['process_config']);
            foreach ($processArr as $line) {
                list($name, $num, $handler) = explode(':', $line);
                if (empty($num)) {
                    continue;
                }
                if (isset($processes[$name])) {
                    continue;
                }
                $processes[$name] = $num;
                $process = new Process($this, trim($name), intval($num), trim($handler));
                $this->instance->addProcess(new \swoole_process([$process, 'run']));
            }
        }
        //单个自定义进程配置
        if (!empty($config['process_name']) && !empty($config['process_num']) && $config['process_num'] > 0) {
            $handler = $config['process_handler'] ?? '';
            if (!isset($processes[$config['process_name']])) {
                $process = new Process($this, $config['process_name'], $config['process_num'], $handler);
                $this->instance->addProcess(new \swoole_process([$process, 'run']));
                $processes[$config['process_name']] = $config['process_num'];
            }
        }
    }

    /**
     * 出现FatalError时，向客户端返回错误信息
     */
    public function onShutdown()
    {
        $error = error_get_last();
        if (empty($error)) {
            return;
        }
        Lock::unlock();
        Log::fatal(LOG_EMERG, ['error_type' => 'fatal'], $error);
        if (Request::type() == Request::HTTP) {
            if (Request::inDebug()) {
                $str = Response::result();
            } else {
                $str = Response::fatalError(500, $error['message'], $error['file'], $error['line']);
            }
            $length = strlen($str);
            $header = "HTTP/1.1 500 Internal Server Error\r\nServer: SQKB-FatalError\r\nContent-Type: text/html\r\nContent-Length: $length\r\n\r\n$str";
            $this->instance->send($this->currentFd, $header);
        } else {
            if (Request::type() == Request::REDIS) {
                $str = "{$error['message']}. {$error['file']}:{$error['line']}";
                $this->instance->send($this->currentFd, Server::format(Server::ERROR, $str));
            } else {
                if (Request::type() == Request::WEBSOCKET) {
                    $str = "{$error['message']}. {$error['file']}:{$error['line']}";
                    $this->instance->push($this->currentFd, $str);
                }
            }
        }
        Response::reset();
    }

    /**
     * 有错误时调用
     */
    public function onError($errno, $errstr, $errfile, $errline)
    {
        try {
            Lock::unlock();
            throw new \Exception;
        } catch (\Exception $exc) {
            // DB连接错误重定向到db_conn_err日志中
            $logFileName = $this->causedByLostConnection($errstr) ? 'db_conn_error' : 'error';
            $errContext = $exc->getTraceAsString();
            $str = sprintf("%s:%d\nerr code:%d\t%s\n%s\n", $errfile, $errline, $errno, $errstr, $errContext);
            $detail = [
                'error_type' => 'error',
                'error_source' => $errfile . ':' . $errline,
                'query' => [],
            ];
            $context = [
                'error_code' => $errno,
                'message' => $errstr,
                'trace' => $exc->getTrace(),
            ];
            Log::$logFileName(Log::errno2LogLevel($errno), $detail, $context);
            if ('/_debug' == Request::uri() && Request::type() == Request::HTTP) {
                var_export([
                    'error_code' => $errno,
                    'message' => $errstr,
                    'file' => $errfile . ':' . $errline,
                    'trace' => $exc->getTrace(),
                ]);
            } else {
                if (Request::type() == Request::REDIS && $this->instance) {
                    $this->instance->send($this->currentFd, '-' . $str);
                }
            }
        }
        return true;
    }

    /**
     * 注册观察者事件及回调
     */
    public static function registerWatcher()
    {
    }

    /**
     * 每个子进程启动时调用
     */
    abstract public function onInit();

    /**
     * Server启动前调用
     */
    abstract public function beforeStart();

    /**
     * 每次请求完后调用
     */
    abstract public function afterRequest();

    /**
     * 每次请求前调用
     */
    public function beforeRequest()
    {
    }

    /**
     * 是否中断此次请求，如果返回布尔值true，则不继续请求
     */
    public function interruptRequest()
    {
        return false;
    }

    /**
     * worker启动前调用
     */
    public function beforeWorkerStart()
    {
    }

    /**
     * 在Server正常结束时发生
     */
    public function onServerShutdown($server)
    {
    }

    /**
     * 注册路由中间件
     * @param $middlewares
     */
    public function registerRouteMiddleware($middlewares)
    {
        foreach ($middlewares as $alias => $middlewareClass) {
            $this->routeMiddlewares->set($alias, ['middleware' => $middlewareClass]);
        }
        Dispatcher::getInstance(CORE_WORK_ROOT)->setRouteMiddlewareObj($this->routeMiddlewares);
    }

    public function registerWsService()
    {
        return WebSocket::getInstance($this->instance);
    }

    /**
     * 识别是否是mysql连接错误报的错
     * @param $errMsg string 报错信息
     * @return bool
     */
    private function causedByLostConnection($errMsg)
    {
        return DbError::causedByLostConnection($errMsg);
    }
}
